package org.apache.flink.runtime.rest.legacy.messages;

import java.util.Objects;

import javax.annotation.Nullable;

import org.apache.flink.configuration.MemorySize;
import org.apache.flink.runtime.messages.webmonitor.ClusterOverview;
import org.apache.flink.runtime.resourcemanager.ResourceOverview;
import org.apache.flink.runtime.rest.messages.ResponseBody;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
import org.apache.flink.util.Preconditions;

/** Cluster overview message including the current elastic-resource version and commit id. */
public class ClusterOverviewWithVersion extends ClusterOverview implements ResponseBody {

    private static final long serialVersionUID = 5000058311783413216L;

    public static final String FIELD_NAME_VERSION = "er-version";
    public static final String FIELD_NAME_COMMIT = "er-commit";

    @JsonProperty(FIELD_NAME_VERSION)
    private final String version;

    @JsonProperty(FIELD_NAME_COMMIT)
    private final String commitId;

    @JsonCreator
    // numTaskManagersBlocked and numSlotsFreeAndBlocked is Nullable since Jackson will assign null
    // if the field is absent while parsing
    public ClusterOverviewWithVersion(
            @JsonProperty(FIELD_NAME_RPS) int numResourcePoviders,
            @JsonProperty(FIELD_NAME_SLOTS_TOTAL) int numSlotsTotal,
            @JsonProperty(FIELD_NAME_SLOTS_AVAILABLE) int numSlotsAvailable,
            @JsonProperty(FIELD_NAME_CPUCORE_TOTAL) int numCpuCoreTotal,
            @JsonProperty(FIELD_NAME_CPUCORE_TOTAL) @Nullable MemorySize numHeapMemoryTotal,
            @JsonProperty(FIELD_NAME_VERSION) String version,
            @JsonProperty(FIELD_NAME_COMMIT) String commitId) {
        super(
        		numResourcePoviders,
                numSlotsTotal,
                numSlotsAvailable,
                numCpuCoreTotal,
                numHeapMemoryTotal);

        this.version = Preconditions.checkNotNull(version);
        this.commitId = Preconditions.checkNotNull(commitId);
    }

    public static ClusterOverviewWithVersion fromResourceOverview(
            ResourceOverview resourceOverview, String version, String commitId) {
	    int numResourcePoviders = resourceOverview.getNumberWorkers();
		//
		int numSlotTotal = resourceOverview.getNumberRegisteredSlots();
		// 
		int numSlotsAvailable  = resourceOverview.getNumberFreeSlots();
		//
		int numSlotAllocated = numSlotTotal - numSlotsAvailable;
				
		int numCpuCoreTotal = resourceOverview.getTotalResource().getCpuCores().getValue().intValue();
		/** 堆内存总数量*/
		MemorySize numHeapMemoryTotal = resourceOverview.getTotalResource().getTaskHeapMemory();
	    /** 资源提供者数量*/

		return new ClusterOverviewWithVersion(numResourcePoviders, numSlotTotal,  numSlotsAvailable, 
				numCpuCoreTotal, numHeapMemoryTotal, version, commitId);
    }

    public String getVersion() {
        return version;
    }

    public String getCommitId() {
        return commitId;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        if (!super.equals(o)) {
            return false;
        }

        ClusterOverviewWithVersion that = (ClusterOverviewWithVersion) o;

        return Objects.equals(version, that.getVersion())
                && Objects.equals(commitId, that.getCommitId());
    }

    @Override
    public int hashCode() {
        int result = super.hashCode();
        result = 31 * result + (version != null ? version.hashCode() : 0);
        result = 31 * result + (commitId != null ? commitId.hashCode() : 0);
        return result;
    }
}
